package com.exercise.module.common.mq.activemq;/**
 * Created by kete on 2017-05-24.
 */

import com.exercise.module.common.config.service.ConfigService;
import com.exercise.module.common.mq.Mq;
import com.exercise.module.common.mq.MqListener;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.ScheduledMessage;
import org.apache.activemq.pool.PooledConnection;
import org.apache.activemq.pool.PooledConnectionFactory;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.jms.*;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 作者：kete
 * 创建时间：2017-05-24 16:44
 * 功能描述：
 * 版本：
 */
@Service
public class activeMqService implements Mq {

    private Logger logger = LoggerFactory.getLogger(activeMqService.class);

    private static PooledConnectionFactory poolFactory;

    private Map<String, List<SubscribeVo>> subscribeVoMap = new HashMap<String, List<SubscribeVo>>();

    @Autowired
    private ConfigService configService;

    @Override
    public String send(String mqCode, String messageBody) {
        return send(mqCode,messageBody,0);
    }

    @Override
    public String send(String mqCode, String messageBody, int deliverTime) {
        logger.debug(">>> mqCode=" + mqCode + " messageBody=" + messageBody);

        try {
            Session session = createSession();


            final String mqTopic = configService.getString("mqTopic");

            Destination destination = null;

            if(mqTopic.equals(Mq.MQ_QUEUE)){
                /** 广播模式 */
                destination = session.createTopic(mqCode);
            }else{
                /** 集群模式 */
                destination = session.createQueue(mqCode);
            }

            MessageProducer producer = session.createProducer(destination);

            producer.setDeliveryMode(DeliveryMode.PERSISTENT);

            TextMessage message = session.createTextMessage(messageBody);

            if (deliverTime > 0) {
                long delay = deliverTime * 1000;
                message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, delay);
            }

            producer.send(message);
            logger.debug("发送结果=" + message);

            //发送结果可以存数据库

            return message.getJMSMessageID();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    @Override
    public void subscribe(String mqCode, MqListener mqListener) {

        logger.info(">>> mqCode=" + mqCode);

        try {

            Connection connection = getConnection();

            Session session =  connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

            final String mqTopic = configService.getString("mqTopic");

            Destination destination = null;

            if(mqTopic.equals(Mq.MQ_QUEUE)){
                /** 广播模式 */
                destination = session.createTopic(mqCode);
            }else{
                /** 集群模式 */
                destination = session.createQueue(mqCode);
            }

            MessageConsumer consumer = session.createConsumer(destination);

            consumer.setMessageListener(new MqListener() {
                @Override
                public void onMessage(Message message) {
                    mqListener.onMessage(message);

                    //修改数据库消息状态
                    //updateMessage(tm.getJMSMessageID(), "已投递成功,消息被处理.", 1);
                }
            });


            SubscribeVo subscribeVo = new SubscribeVo();
            subscribeVo.setConnection(connection);
            subscribeVo.setSession(session);

            List subscribeVoList = subscribeVoMap.get(mqCode);
            if (subscribeVoList == null) {
                subscribeVoList = new ArrayList();
                subscribeVoMap.put(mqCode, subscribeVoList);
            }
            subscribeVoList.add(subscribeVo);

        } catch (JMSException e) {
            e.printStackTrace();
        }

    }

    @Override
    public void unSubscribe(String mqCode) {


        try {
            List<SubscribeVo> subscribeVoList = subscribeVoMap.get(mqCode);

            if (subscribeVoList != null) {
                for (SubscribeVo subscribeVo : subscribeVoList) {
                    subscribeVo.getSession().close();
                    subscribeVo.getConnection().close();
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    public Session createSession() throws JMSException {

        PooledConnectionFactory poolFactory = getPooledConnectionFactory();
        PooledConnection pooledConnection = (PooledConnection) poolFactory.createConnection();

        return pooledConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    }

    public Connection getConnection() {

        Connection connection = null;
        try {
            ActiveMQConnectionFactory factory = null;
            String mqAddr = configService.getString("mqAddr");
            String mqUsernamePassword = configService.getString("mqUsernamePassword");
            logger.info("mqAddr:" + mqAddr + " mqUsernamePassword:" + mqUsernamePassword);
            if (StringUtils.isNotBlank(mqUsernamePassword)) {
                String[] mqInfo = StringUtils.split(mqUsernamePassword, ":");
                String userName = mqInfo[0];
                String password = mqInfo[1];
                factory = new ActiveMQConnectionFactory(userName, password, mqAddr);
            } else {
                factory = new ActiveMQConnectionFactory(mqAddr);
            }

            connection = factory.createConnection();
            connection.start();

            return connection;
        } catch (Exception e) {
            e.printStackTrace();
            close(connection);
            return null;
        }
    }


    public void close(Connection connection) {
        try {
            if (connection != null)
                connection.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    private synchronized PooledConnectionFactory getPooledConnectionFactory() {

        if (poolFactory != null) return poolFactory;

        logger.info("创建新的mq连接池");

        ActiveMQConnectionFactory factory = null;
        //tcp://127.0.0.1:61616
        String mqAddr = configService.getString("mqAddr");
        //zcjb_mq:zcjb_2017
        String mqUsernamePassword = configService.getString("mqUsernamePassword");
        if (StringUtils.isNotBlank(mqUsernamePassword)) {
            String[] mqInfo = StringUtils.split(mqUsernamePassword, ":",2);
            String userName = mqInfo[0];
            String password = mqInfo[1];
            factory = new ActiveMQConnectionFactory(userName, password, mqAddr);
        } else {
            factory = new ActiveMQConnectionFactory(mqAddr);
        }

        poolFactory = new PooledConnectionFactory(factory);
        /** 池中借出的对象的最大数目 */
        poolFactory.setMaxConnections(100);
        poolFactory.setMaximumActiveSessionPerConnection(50);
        /** 后台对象清理时，休眠时间超过了3000毫秒的对象为过期 */
        poolFactory.setTimeBetweenExpirationCheckMillis(3000);
        logger.info(" 创建新的mq连接池 success");
        return poolFactory;
    }
}
